package com.tang.transform;

import com.tang.bean.WaterSensor;
import com.tang.functions.FlatMapFunctionImpl;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * 扁平转换demo
 *
 * @author tang
 * @since 2023/6/7 10:36
 */
public class FlatmapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 1),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s4", 4L, 4)
        );

        /*
         * 对于s1数据，1进1出
         * 对于s2数据，1进2出
         * 对于s3数据，1进0出
         * 
         * map怎么控制一进一出？
         * => 使用return
         * 
         * flatMap怎么控制一进多出？
         * => 通过Collector来输出，调用几次就输出几条
         */
        source.flatMap(new FlatMapFunction<WaterSensor, String>() {
            @Override
            public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
                if ("s1".equals(value.getId())) {
                    // 如果是s1，输出vc
                    out.collect(value.getVc().toString());
                } else if ("s2".equals(value.getId())){
                    // 如果是s2，输出ts与vc
                    out.collect(value.getTs().toString());
                    out.collect(value.getVc().toString());
                }
            }
            
        }).print();
        
        source.flatMap((FlatMapFunction<WaterSensor, String>) (value, out) -> {
            if ("s1".equals(value.getId())) {
                // 如果是s1，输出vc
                out.collect(value.getVc().toString());
            } else if ("s2".equals(value.getId())){
                // 如果是s2，输出ts与vc
                out.collect(value.getTs().toString());
                out.collect(value.getVc().toString());
            }
        }).returns(Types.STRING).print();

        source.flatMap(new FlatMapFunctionImpl()).print();

        env.execute();
    }
    
}
